package com.spx.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * create by undeRdoG on  2022-05-10  14:37
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class CommonApiTest {

    public static void main(String[] args) {
        /*  依赖 StreamExecutionEnvironment 来创建表的执行环境
        *StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        *
        *StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        **/



        //完全摆脱DataStream的env，单独的来创建表执行环境

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(environmentSettings);


        // 基于 blink版本的 批处理的表执行环境
       /* EnvironmentSettings environmentSettings1 = EnvironmentSettings.newInstance()
                .inBatchMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tableEnv1 = TableEnvironment.create(environmentSettings1);*/


        //=================================================创建表================================================================//

        // 创建读入表
        String createDDL = "CREATE TABLE clickTable (" +
                "`user` STRING, " +
                "url STRING, " +
                "ts BIGINT" +
                ") WITH (" +
                " 'connector' = 'filesystem' , " +
                " 'path' = 'input/clicks.txt' , " +
                " 'format' = 'csv'"
                + ")";

        // 创建输出表
        String createOutDDL = "CREATE TABLE outTable (" +
                "`user` STRING, " +
                "url STRING " +
                ") WITH (" +
                " 'connector' = 'filesystem' , " +
                " 'path' = 'output' , " +
                " 'format' = 'csv'"
                + ")";

        // 创建一张用于控制台打印的表
        String createPrintDDL = "CREATE TABLE printTable (" +
                "`user` STRING, " +
                "url STRING " +
                ") WITH (" +
                " 'connector' = 'print'  "
                + ")";


        tableEnv.executeSql(createDDL);
        tableEnv.executeSql(createOutDDL);
        tableEnv.executeSql(createPrintDDL);

        Table resultTable = tableEnv.sqlQuery("select `user`,url from clickTable where user = 'Bob'");

        resultTable.executeInsert("outTable");
        //resultTable.executeInsert("printTable");

        // 执行聚合计算
        Table aggResult = tableEnv.sqlQuery("select `user`,count(*) as `cnt` from clickTable group by `user`");

        // 创建一张用于聚合打印的输出表
        String createPrintAggDDL = "CREATE TABLE printAggTable (" +
                "`user` STRING, " +
                "`count` BIGINT " +
                ") WITH (" +
                " 'connector' = 'print'  "
                + ")";
        tableEnv.executeSql(createPrintAggDDL);
        aggResult.executeInsert("printAggTable");


    }

}
